home *** CD-ROM | disk | FTP | other *** search
/ Mac Easy 2010 May / Mac Life Ubuntu.iso / casper / filesystem.squashfs / usr / share / python-support / python-gdata / atom / mock_http.py < prev    next >
Encoding:
Python Source  |  2008-09-09  |  4.4 KB  |  133 lines
  1. #!/usr/bin/python
  2. #
  3. # Copyright (C) 2008 Google Inc.
  4. #
  5. # Licensed under the Apache License, Version 2.0 (the "License");
  6. # you may not use this file except in compliance with the License.
  7. # You may obtain a copy of the License at
  8. #
  9. #      http://www.apache.org/licenses/LICENSE-2.0
  10. #
  11. # Unless required by applicable law or agreed to in writing, software
  12. # distributed under the License is distributed on an "AS IS" BASIS,
  13. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. # See the License for the specific language governing permissions and
  15. # limitations under the License.
  16.  
  17.  
  18. __author__ = 'api.jscudder (Jeff Scudder)'
  19.  
  20.  
  21. import atom.http_interface
  22. import atom.url
  23.  
  24.  
  25. class Error(Exception):
  26.   pass
  27.  
  28.  
  29. class NoRecordingFound(Error):
  30.   pass
  31.  
  32.  
  33. class MockRequest(object):
  34.   """Holds parameters of an HTTP request for matching against future requests.
  35.   """
  36.   def __init__(self, operation, url, data=None, headers=None):
  37.     self.operation = operation
  38.     if isinstance(url, (str, unicode)):
  39.       url = atom.url.parse_url(url)
  40.     self.url = url
  41.     self.data = data
  42.     self.headers = headers
  43.  
  44.  
  45. class MockResponse(atom.http_interface.HttpResponse):
  46.   """Simulates an httplib.HTTPResponse object."""
  47.   def __init__(self, body=None, status=None, reason=None, headers=None):
  48.     if body and hasattr(body, 'read'):
  49.       self.body = body.read()
  50.     else:
  51.       self.body = body
  52.     if status is not None:
  53.       self.status = int(status)
  54.     else:
  55.       self.status = None
  56.     self.reason = reason
  57.     self._headers = headers or {}
  58.  
  59.   def read(self):
  60.     return self.body
  61.  
  62.  
  63. class MockHttpClient(atom.http_interface.GenericHttpClient):
  64.   def __init__(self, headers=None, recordings=None, real_client=None):
  65.     """An HttpClient which responds to request with stored data.
  66.  
  67.     The request-response pairs are stored as tuples in a member list named
  68.     recordings.
  69.  
  70.     The MockHttpClient can be switched from replay mode to record mode by
  71.     setting the real_client member to an instance of an HttpClient which will
  72.     make real HTTP requests and store the server's response in list of 
  73.     recordings.
  74.     
  75.     Args:
  76.       headers: dict containing HTTP headers which should be included in all
  77.           HTTP requests.
  78.       recordings: The initial recordings to be used for responses. This list
  79.           contains tuples in the form: (MockRequest, MockResponse)
  80.       real_client: An HttpClient which will make a real HTTP request. The 
  81.           response will be converted into a MockResponse and stored in 
  82.           recordings.
  83.     """
  84.     self.recordings = recordings or []
  85.     self.real_client = real_client
  86.     self.headers = headers or {}
  87.  
  88.   def add_response(self, response, operation, url, data=None, headers=None):
  89.     """Adds a request-response pair to the recordings list.
  90.     
  91.     After the recording is added, future matching requests will receive the
  92.     response.
  93.     
  94.     Args:
  95.       response: MockResponse
  96.       operation: str
  97.       url: str
  98.       data: str, Currently the data is ignored when looking for matching
  99.           requests.
  100.       headers: dict of strings: Currently the headers are ignored when
  101.           looking for matching requests.
  102.     """
  103.     request = MockRequest(operation, url, data=data, headers=headers)
  104.     self.recordings.append((request, response))
  105.  
  106.   def request(self, operation, url, data=None, headers=None):
  107.     """Returns a matching MockResponse from the recordings.
  108.     
  109.     If the real_client is set, the request will be passed along and the 
  110.     server's response will be added to the recordings and also returned. 
  111.  
  112.     If there is no match, a NoRecordingFound error will be raised.
  113.     """
  114.     if self.real_client is None:
  115.       if isinstance(url, (str, unicode)):
  116.         url = atom.url.parse_url(url)
  117.       for recording in self.recordings:
  118.         if recording[0].operation == operation and recording[0].url == url:
  119.           return recording[1]
  120.       raise NoRecordingFound('No recodings found for %s %s' % (
  121.           operation, url))
  122.     else:
  123.       # There is a real HTTP client, so make the request, and record the 
  124.       # response.
  125.       response = self.real_client.request(operation, url, data=data, 
  126.           headers=headers)
  127.       # TODO: copy the headers
  128.       stored_response = MockResponse(body=response, status=response.status,
  129.           reason=response.reason)
  130.       self.add_response(stored_response, operation, url, data=data, 
  131.           headers=headers)
  132.       return stored_response
  133.